package io.kiki.sba.registry.server.clientInterface.remoting.handler;

import io.kiki.sba.registry.api.Channel;
import io.kiki.sba.registry.common.model.Node.NodeType;
import io.kiki.sba.registry.common.model.dataserver.DatumVersion;
import io.kiki.sba.registry.common.model.sessionserver.DataChangeRequest;
import io.kiki.sba.registry.server.clientInterface.bootstrap.Config;
import io.kiki.sba.registry.server.clientInterface.bootstrap.ExecutorManager;
import io.kiki.sba.registry.server.clientInterface.cache.CacheService;
import io.kiki.sba.registry.server.clientInterface.push.FirePushService;
import io.kiki.sba.registry.server.clientInterface.push.PushSwitchService;
import io.kiki.sba.registry.server.clientInterface.push.TriggerPushContext;
import io.kiki.sba.registry.server.clientInterface.store.Subscribers;
import io.kiki.sba.registry.server.shared.remoting.AbstractClientHandler;
import io.kiki.sba.registry.server.shared.remoting.RemotingHelper;
import io.kiki.sba.registry.util.StringUtil;
import lombok.NonNull;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;

import java.util.Collections;
import java.util.Map;
import java.util.concurrent.Executor;


public class DataChangeRequestHandler extends AbstractClientHandler<DataChangeRequest> {

    private static final Logger logger = LoggerFactory.getLogger(DataChangeRequestHandler.class);
    /**
     * store subscribers
     */
    @Autowired
    Subscribers sessionSubscribers;

    @Autowired
    Config config;

    @Autowired
    ExecutorManager executorManager;

    @Autowired
    FirePushService firePushService;

    @Autowired
    PushSwitchService pushSwitchService;

    @Autowired
    CacheService sessionCacheService;

    @Override
    protected NodeType getConnectNodeType() {
        return NodeType.data_store;
    }

    @Override
    public Executor getExecutor() {
        return executorManager.getDataChangeRequestExecutor();
    }

    @Override
    public void checkParam(@NonNull DataChangeRequest request) {

        StringUtil.checkNotBlank(request.getDataCenter(), "request.dataCenter");
        StringUtil.nullValue(request.getDataInfoIdToDatumVersionMap(), "request.dataInfoIds");
    }

    @Override
    public Object doHandle(Channel channel, DataChangeRequest dataChangeRequest) {
        if (!pushSwitchService.canLocalDataCenterPush()) {
            return null;
        }
        final String dataNode = RemotingHelper.getRemoteHostAddress(channel);
        final String dataCenter = dataChangeRequest.getDataCenter();
        final long changeTimestamp = System.currentTimeMillis();
        for (Map.Entry<String, DatumVersion> entry : dataChangeRequest.getDataInfoIdToDatumVersionMap().entrySet()) {

            final String dataInfoId = entry.getKey();
            if (!pushSwitchService.canPushMultiDataCenter(Collections.singleton(dataCenter))) {
                continue;
            }
            final DatumVersion datumVersion = entry.getValue();
            Subscribers.SubscriberVersionCheck check = sessionSubscribers.checkInterestVersion(dataCenter, dataInfoId, datumVersion.getValue());
            if (!check.interested) {
                if (check != Subscribers.SubscriberVersionCheck.NoSub) {
                    // log exclude NoSub
                    logger.info("[SkipChange]{},{}, ver={}, {}", dataInfoId, dataCenter, datumVersion, check);
                }
                continue;
            }
            final TriggerPushContext changeCtx = new TriggerPushContext(dataCenter, datumVersion.getValue(), dataNode, changeTimestamp, dataChangeRequest.getTimes());
            firePushService.fireOnChange(dataInfoId, changeCtx);
        }
        return null;
    }

    @Override
    public Class interest() {
        return DataChangeRequest.class;
    }
}
